package cn.itcast.function;

import cn.itcast.bean.SectorBean;
import cn.itcast.bean.StockBean;
import cn.itcast.util.DbUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

//秒级和分时写在一起了
public class SectorWindowFunction extends RichAllWindowFunction<StockBean, SectorBean, TimeWindow> {
    //map<板块,List<个股，一行个股数据>>
    private Map<String, List<Map<String, Object>>> sectorStockMap=null;
    //最近交易日日k(上一个交易日）
    private Map<String, Map<String, Object>> sectorKlineMap = null;

    private MapState<String, SectorBean> sectorMs=null;
    BigDecimal basePrice = new BigDecimal(1000);
    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         *   1. 初始化数据：板块对应关系、最近交易日日K（上一交易日）
         *   2. 定义状态MapState<String, SectorBean >:上一个窗口板块数据
         *   3. 初始化基准价（1000）
         */
        //1.查询板块个股对应关系表--只查询沪市个股
        String sql = "select * from bdp_sector_stock";
        sectorStockMap = DbUtil.queryForGroup("sector_code", sql);
        //查询最近交易日日K(上一交易日）
        String sqlKline = "select * from bdp_quot_sector_kline_day where trade_date = (select last_trade_date from tcc_date where trade_date = curdate())";
        sectorKlineMap = DbUtil.query("sector_code", sqlKline);

        //定义状态MapState<String,SectorBean>:上一个窗口板块数据
        sectorMs = getRuntimeContext().getMapState(new MapStateDescriptor<String, SectorBean>(
                "sectorMs", String.class, SectorBean.class
        ));
    }

    @Override
    public void apply(TimeWindow timeWindow, Iterable<StockBean> iterable, Collector<SectorBean> collector) throws Exception {
        /**
         * 1.循环窗口内个股数据并缓存
         * 2.轮询板块对应关系表下的个股数据、并获取指定板块下的个股
         * 3.初始化全部数据
         * 4.轮询板块下的个股数据，并获取板块名称、个股代码、个股流通股本和前一日板块总流通市值
         * 5.计算：开盘价/收盘价累计流通市值、累计（交易量、交易金额）（注意：是个股累计值，获取缓存个股数据）
         *    累计个股开盘流通市值 = SUM(板块下的个股开盘价*个股流通股本)
         *    累计个股收盘流通市值 = SUM(板块下的个股收盘价*个股流通股本)
         * 6.判断是否是首日上市，并计算板块开盘价和收盘价
         *    板块开盘价格 = 板块前收盘价*当前板块以开盘价计算的总流通市值/当前板块前一交易日板块总流通市值
         *    板块当前价格 = 板块前收盘价*当前板块以收盘价计算的总流通市值/当前板块前一交易日板块总流通市值
         * 7.初始化板块高低价
         * 8.获取上一个窗口板块数据（高、低、成交量、成交金额）
         * 9.计算最高价和最低价（前后窗口比较）
         * 10.计算分时成交量和成交金额
         * 11.开盘价与高低价比较
         * 12.封装结果数据
         * 13.缓存当前板块数据
         */
        //todo 1.循环窗口内个股数据并缓存
        Map<String, StockBean> map = new HashMap<>();
        for (StockBean value : iterable) {
            map.put(value.getSecCode(),value);
        }

        //todo 2.轮询板块对应关系表下的个股数据、
        for (String sectorCode : sectorStockMap.keySet()) {
            //获取指定板块下的个股
            List<Map<String, Object>> list = sectorStockMap.get(sectorCode);

            //todo 3.初始化全部数据
            Long eventTime = 0L;
            String sectorName = null;
            BigDecimal preClosePrice = new BigDecimal(0);
            BigDecimal openPrice = new BigDecimal(0);
            BigDecimal highPrice = new BigDecimal(0);
            BigDecimal lowPrice = new BigDecimal(0);
            BigDecimal closePrice = new BigDecimal(0);
            Long tradeVol = 0L; //分时成交量
            Long tradeAmt = 0L;
            Long tradeVolDay = 0L; //总成交量
            Long tradeAmtDay = 0L;
            Long tradeTime = 0L;


            //累计个股开盘流通市值
            BigDecimal totalOpenNegoCap = new BigDecimal(0);
            //累计个股收盘流通市值
            BigDecimal totalCloseNegoCap = new BigDecimal(0);
            //前一日板块总流通市值
            BigDecimal preSectorNegoCap = new BigDecimal(0);
            //todo 4.轮询板块下的个股数据，并获取板块名称、个股代码、个股流通股本和前一日板块总流通市值
            for (Map<String, Object> stockMap : list) {
                sectorName = stockMap.get("sector_name").toString();
                String secCode = stockMap.get("sec_code").toString();
                BigDecimal negoCap = new BigDecimal(stockMap.get("nego_cap").toString());
                //前一日板块总流通市值
                preSectorNegoCap = new BigDecimal(stockMap.get("pre_sector_nego_cap").toString());
                //todo 5.计算：开盘价/收盘价累计流通市值、累计（交易量、交易金额）（注意：是个股累计值，获取缓存个股数据）
                StockBean stockBean = map.get(secCode);
                if(stockBean!=null){ //说明实时流中的个股数据一定存在板块个股对应关系表中
                    eventTime = stockBean.getEventTime();
                    tradeTime = stockBean.getTradeTime();
                    //板块下的所以个股累计(交易量和交易金额)
                    tradeVolDay += stockBean.getTradeVolDay(); //板块总交易量
                    tradeAmtDay += stockBean.getTradeAmtDay();  //板块总交易金额

                    //个股开盘流通市值=板块下的个股开盘价*个股流通股本
                    BigDecimal singleStockOpen = stockBean.getOpenPrice().multiply(negoCap).setScale(2, RoundingMode.HALF_UP);
                    //累计个股开盘流通市值 = SUM(板块下的个股开盘价*个股流通股本)
                    totalOpenNegoCap = totalOpenNegoCap.add(singleStockOpen);
                    //个股收盘流通市值=板块下的个股收盘价*个股流通股本
                    BigDecimal singleStockClose = stockBean.getClosePrice().multiply(negoCap).setScale(2, RoundingMode.HALF_UP);
                    //累计个股收盘流通市值 = SUM(板块下的个股收盘价*个股流通股本)
                    totalCloseNegoCap = totalCloseNegoCap.add(singleStockClose);
                }
            }
            //todo 6.判断是否是首日上市，并计算板块开盘价和收盘价
            //        板块开盘价格 = 板块前收盘价*当前板块以开盘价计算的总流通市值/当前板块前一交易日板块总流通市值
            //        板块当前价格 = 板块前收盘价*当前板块以收盘价计算的总流通市值/当前板块前一交易日板块总流通市值
            if(sectorKlineMap==null || sectorKlineMap.get(sectorCode)==null){//说明板块首日上市
                preClosePrice= basePrice;
                openPrice = (preClosePrice.multiply(totalOpenNegoCap)).divide(preSectorNegoCap,2,RoundingMode.HALF_UP);
                closePrice = (preClosePrice.multiply(totalCloseNegoCap)).divide(preSectorNegoCap,2,RoundingMode.HALF_UP);
            }else {
                Map<String, Object> sectorMap = sectorKlineMap.get(sectorCode);
                if(sectorMap!=null){//说明此版块已经上市，并且前一交易日已经有数据
                    //前收盘价=前一交易日的收盘价
                    preClosePrice = new BigDecimal(sectorMap.get("close_price").toString());
                    openPrice = (preClosePrice.multiply(totalOpenNegoCap)).divide(preSectorNegoCap,2,RoundingMode.HALF_UP);
                    closePrice = (preClosePrice.multiply(totalCloseNegoCap)).divide(preSectorNegoCap,2,RoundingMode.HALF_UP);
                }
            }
            //todo 7.初始化板块高低价
            highPrice = closePrice;
            lowPrice = closePrice;
            //todo 8.获取上一个窗口板块数据（高、低、成交量、成交金额）
            SectorBean sectorBeanLast = sectorMs.get(sectorCode);
            if(sectorBeanLast!=null){
                BigDecimal highPriceLast = sectorBeanLast.getHighPrice();
                BigDecimal lowPriceLast = sectorBeanLast.getLowPrice();
                Long tradeVolDayLast = sectorBeanLast.getTradeVolDay();
                Long tradeAmtDayLast = sectorBeanLast.getTradeAmtDay();
                //todo 9.计算分时成交量和成交金额
                tradeVol = tradeVolDay-tradeVolDayLast;//分时成交量和成交金额
                tradeAmt = tradeAmtDay-tradeAmtDayLast;
                //todo 10.计算最高价和最低价（前后窗口比较）
                if(highPrice.compareTo(highPriceLast)==-1){ // <
                    highPrice = highPriceLast;
                }
                if(lowPrice.compareTo(lowPriceLast)==1){ //>
                    lowPrice = lowPriceLast;
                }
            }
            //todo 11.开盘价与高低价比较(解决高开低收）
            if(openPrice.compareTo(highPrice)==1){ //>
                highPrice = openPrice;
            }
            if(openPrice.compareTo(lowPrice)==-1){ //<
                lowPrice = openPrice;
            }
            //todo 12.封装结果数据
            SectorBean sectorBean = new SectorBean();
            sectorBean.setEventTime(eventTime);
            sectorBean.setSectorCode(sectorCode);
            sectorBean.setSectorName(sectorName);
            sectorBean.setPreClosePrice(preClosePrice);
            sectorBean.setOpenPrice(openPrice);
            sectorBean.setHighPrice(highPrice);
            sectorBean.setLowPrice(lowPrice);
            sectorBean.setClosePrice(closePrice);
            sectorBean.setTradeVol(tradeVol);
            sectorBean.setTradeAmt(tradeAmt);
            sectorBean.setTradeVolDay(tradeVolDay);
            sectorBean.setTradeAmtDay(tradeAmtDay);
            sectorBean.setTradeTime(tradeTime);
            //收集
            collector.collect(sectorBean);
            //todo 13.缓存当前板块数据
            sectorMs.put(sectorCode,sectorBean);
        }
    }
}
